package com.yuxl.common.core.utils;

import com.yuxl.common.core.domain.socket.Connects;
import io.lettuce.core.RedisClient;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.lettuce.core.pubsub.api.sync.RedisPubSubCommands;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.socket.WebSocketSession;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;
import java.util.Set;

/**
 * 集群WebSocket推送工具
 * @author yuxl
 */
@Component
public class SocketUtil {
    private static ReactiveRedisTemplate<String, String> reactiveRedisTemplate;
    private static final String WS_PUSH_USER_PREFIX = "ws:push:user:";
    private static final String WS_PUSH_BROADCAST = "ws:push:all";

    @Value("${spring.data.redis.host}")
    private String redisHost;
    @Value("${spring.data.redis.port}")
    private int redisPort;
    @Value("${spring.data.redis.password:}")
    private String redisPassword;

    @Resource(name = "reactiveRedisTemplate")
    public void setRedisTemplate(ReactiveRedisTemplate<String, String> reactiveRedisTemplate) {
        SocketUtil.reactiveRedisTemplate = reactiveRedisTemplate;
    }

    @PostConstruct
    public void subscribe() {
        // 订阅广播频道
        reactiveRedisTemplate.listenToChannel(WS_PUSH_BROADCAST)
                .flatMap(msg -> sendToAllLocal(msg.getMessage()))
                .subscribe();

        // 使用Lettuce原生API订阅所有用户频道，支持密码
        String redisUrl;
        if (redisPassword != null && !redisPassword.isEmpty()) {
            redisUrl = "redis://:" + redisPassword + "@" + redisHost + ":" + redisPort;
        } else {
            redisUrl = "redis://" + redisHost + ":" + redisPort;
        }
        RedisClient client = RedisClient.create(redisUrl);
        StatefulRedisPubSubConnection<String, String> connection = client.connectPubSub();
        RedisPubSubCommands<String, String> sync = connection.sync();
        sync.psubscribe(WS_PUSH_USER_PREFIX + "*");
        connection.addListener(new io.lettuce.core.pubsub.RedisPubSubListener<>() {
            @Override
            public void message(String channel, String message) {
                // 普通订阅消息
            }
            @Override
            public void message(String pattern, String channel, String message) {
                // 模式订阅消息
                if (channel.startsWith(WS_PUSH_USER_PREFIX)) {
                    String username = channel.substring(WS_PUSH_USER_PREFIX.length());
                    sendToUserLocal(username, message).subscribe();
                }
            }
            public void subscribed(String channel, long count) {}
            public void psubscribed(String pattern, long count) {}
            public void unsubscribed(String channel, long count) {}
            public void punsubscribed(String pattern, long count) {}
        });
    }

    /**
     * 给指定用户的所有连接发送消息（集群）
     */
    public static <T> Mono<Long> sendToUser(String username, T message) {
        return reactiveRedisTemplate.convertAndSend(WS_PUSH_USER_PREFIX + username, message.toString());
    }

    /**
     * 给所有连接发送消息（集群）
     */
    public static <T> Mono<Long> sendToAll(T message) {
        return reactiveRedisTemplate.convertAndSend(WS_PUSH_BROADCAST, message.toString());
    }

    // 本地推送实现
    private static <T> Flux<Void> sendToUserLocal(String username, T message) {
        List<Mono<Void>> list = Connects.getSessions(username).stream().map(session ->
                session.send(Mono.just(session.textMessage(message.toString())))
        ).toList();
        return Flux.merge(list);
    }
    // 本地推送实现
    private static <T> Flux<Void> sendToAllLocal(T message) {
        List<Mono<Void>> list = Connects.all().stream().map(session ->
                session.send(Mono.just(session.textMessage(message.toString())))
        ).toList();
        return Flux.merge(list);
    }
    // 给指定连接发送消息
    public static <T> Mono<Void> sendToSession(WebSocketSession session, T message) {
        return session.send(Mono.just(session.textMessage(message.toString())));
    }
}
